package com.game.gateway.service;

import com.game.gateway.channel.ServerChannelGroup;
import com.game.framework.message.stream.GatewayMessageHeader;
import com.game.framework.message.stream.ServiceMessageHeader;
import com.game.framework.message.stream.TransferMessage;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.env.Environment;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;

/**
 *  接收业务服务广播返回的消息，并发送到客户端
 */
@Slf4j
@Service
//监听服务广播的消息，使用一个固定的topic即可。
@RocketMQMessageListener(topic = "${mq.topic}",consumerGroup = "${mq.consumer.group}", messageModel = MessageModel.BROADCASTING )
public class ReceiverBroadcastMessageService implements RocketMQListener<MessageExt> {

	protected ServerChannelGroup serverChannelGroup;

	@Autowired
	protected Environment environment;

	private String topic;

	@PostConstruct
	public void init() {
		RocketMQMessageListener rocketMQMessageListener = this.getClass().getAnnotation(RocketMQMessageListener.class);
		if (rocketMQMessageListener == null) {
			log.error("没有添加广播消息监听");
			return;
		}

		topic = environment.resolveRequiredPlaceholders(rocketMQMessageListener.topic());
		String consumerGroup = environment.resolveRequiredPlaceholders(rocketMQMessageListener.consumerGroup());
		log.info("广播消息监听 topic:{}, group:{}",topic, consumerGroup);
	}

	@Override
	public void onMessage(MessageExt message) {
		log.info("ReceiverBroadcastMessageService receive message : {}", message.toString());
		byte[] data = message.getBody();
		if (data != null) {
			try {
				ServiceMessageHeader header = new ServiceMessageHeader();
				ByteBuf byteBuf = Unpooled.wrappedBuffer(data);
				header.read(byteBuf);
				long serverId = header.getToGameServerId();
				Channel channel = serverChannelGroup.getChannel(serverId);// 根据userId找到这个客户端的连接Channel
				if (channel != null) {
					GatewayMessageHeader gatewayMessageHeader = new GatewayMessageHeader();
					//转化为网关包头数据
					BeanUtils.copyProperties(header, gatewayMessageHeader);
					ByteBuf bodyBuf = null;
					if(byteBuf.isReadable()) {
						byte[] bytes = new byte[byteBuf.readableBytes()];
						byteBuf.readBytes(bytes);
						bytes = compress(bytes,gatewayMessageHeader);
						bytes = encrypt(bytes,gatewayMessageHeader);
						bodyBuf = Unpooled.wrappedBuffer(bytes);
					}
					TransferMessage transferMessage = new TransferMessage(gatewayMessageHeader, bodyBuf);
					channel.writeAndFlush(transferMessage);// 给客户端返回消息
				}
				log.debug("返回业务消息响应：{}", header);
			} catch (Throwable e) {
				log.error("接业务收消息异常", e);
			}
		}
	}

	private static byte[] compress(byte[] data,GatewayMessageHeader header) {
		//TODO 在这里对消息进行是否压缩的判断
		return data;
	}
	private static byte[] encrypt(byte[] data,GatewayMessageHeader header) {
		//TODO 在这里对消息进行是否加密的判断
		return data;
	}
}
